package com.easypay.reactive8583;

import com.easypay.reactive8583.iso.MessageFactory;
import com.easypay.reactive8583.netty.pipeline.CompositeIsoMessageHandler;
import com.easypay.reactive8583.netty.pipeline.EchoMessageListener;
import com.solab.iso8583.IsoMessage;
import io.netty.bootstrap.AbstractBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.atomic.AtomicReference;

import static java.lang.Boolean.parseBoolean;

/**
 * @author pandans
 * @date 2021/03/04 14:44
 * @Description:
 */
public abstract class AbstractIso8583Connector<C extends ConnectorConfiguration, B extends AbstractBootstrap<B, ?>, M extends IsoMessage> {
    protected Logger logger = LoggerFactory.getLogger(this.getClass());
    private CompositeIsoMessageHandler<M> messageHandler;
    protected MessageFactory<M> isoMessageFactory;
    private AtomicReference<Channel> channelRef = new AtomicReference<>();
    protected C configuration;
    public ConnectorConfigurer<C, B> configurer;
    protected EventLoopGroup bossEventLoopGroup;
    protected EventLoopGroup workerEventLoopGroup;
    protected B bootstrap;

    public AbstractIso8583Connector(C configuration, MessageFactory<M> isoMessageFactory, CompositeIsoMessageHandler<M> messageHandler) {
        this.isoMessageFactory = isoMessageFactory;
        this.configuration = configuration;
        this.messageHandler = messageHandler;
        initListener();
    }

    public AbstractIso8583Connector(C configuration, MessageFactory<M> isoMessageFactory) {
        super();
        this.configuration = configuration;
        this.messageHandler = new CompositeIsoMessageHandler<>();
        this.isoMessageFactory = isoMessageFactory;
        initListener();
    }

    public void initListener() {
        if (configuration.shouldAddEchoMessageListener()) {
            messageHandler.addListener(new EchoMessageListener(isoMessageFactory));
        }
    }

    public CompositeIsoMessageHandler<M> getMessageHandler(){
        return this.messageHandler;
    }

    public void addMessageListener(IsoMessageListener<M> handler) {
        messageHandler.addListener(handler);
    }

    public void removeMessageListener(IsoMessageListener<M> handler) {
        messageHandler.removeListener(handler);
    }

    /**
     * Making connector ready to create a connection / bind to port.
     * Creates a Bootstrap
     *
     * @see AbstractBootstrap
     */
    public void init() {
        logger.info("Initializing");
        bossEventLoopGroup = createBossEventLoopGroup();
        workerEventLoopGroup = createWorkerEventLoopGroup();
        bootstrap = createBootstrap();
    }

    public void shutdown() {
        workerEventLoopGroup.shutdownGracefully();
        bossEventLoopGroup.shutdownGracefully();
    }

    protected void configureBootstrap(B bootstrap) {
        bootstrap.option(
                ChannelOption.TCP_NODELAY,
                parseBoolean(
                        System.getProperty(
                                "nfs.rpc.tcp.nodelay", "true"
                        )
                )
        )
                .option(ChannelOption.AUTO_READ, true);
        if(configurer != null) {
            configurer.configureBootstrap(bootstrap, configuration);
        }

    }
    public void setConfigurer(ConnectorConfigurer<C, B> configurer) {
        this.configurer = configurer;
    }

    public MessageFactory<M> getIsoMessageFactory() {
        return isoMessageFactory;
    }

    protected abstract B createBootstrap();

    protected EventLoopGroup createBossEventLoopGroup() {
        return new NioEventLoopGroup();
    }

    protected EventLoopGroup createWorkerEventLoopGroup() {
        NioEventLoopGroup group = new NioEventLoopGroup(configuration.getWorkerThreadsCount());
        logger.debug(
                "Created worker EventLoopGroup with {} executor threads",
                group.executorCount()
        );
        return group;
    }
    public Channel getChannel(){
        return channelRef.get();
    }
    public void setChannel(Channel channel) {
        channelRef.set(channel);
    }
}
